package com.wanke.publish;

import cn.beepower.domain.bus.MqttDataBus;
import cn.beepower.domain.bus.ThreadMode;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wanke.domain.po.msg.CloudMeasureValue;
import com.wanke.domain.po.msg.CloudMeasureValues;
import com.wanke.scada.mqtt.MessageKit;
import com.wanke.constant.MqttTopicConstants;
import com.wanke.constant.ProtocolConstants;
import com.wanke.domain.DataTransferDTO;
import com.wanke.domain.po.Measurement;
import com.wanke.domain.po.Measurements;
import io.vertx.core.json.Json;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.wanke.service.MeasurementService;

import java.time.Instant;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * mqtt 量测上传程序
 * @author diaoshili
 * @date 2021-02-02
 */
public class MqttRemoteMeasurePublish extends RemoteMeasurePublish {

    private static final ProtocolConstants protocol = ProtocolConstants.MQTT;
    private static final Logger logger = LoggerFactory.getLogger(MqttRemoteMeasurePublish.class);
    public static  AtomicBoolean  bridgeBrokerConnected = new AtomicBoolean(true);
    public static AtomicLong systemHeartBeatTime = new AtomicLong(0);
    private MqttDataBus dataBus;
    private  MeasurementService measurementService;
    private  Long stationCode;
    private  Long timeOut;
    private  int pollSize = 500;
    private Long pointNumber;
    private Long frequency;
    private String remoteClientId;

    private ObjectMapper objectMapper = new ObjectMapper();

    public MqttRemoteMeasurePublish(MqttDataBus dataBus,
                                    MeasurementService measurementService,
                                    Long stationCode,
                                    Long timeOut,
                                    int pollSize,
                                    Long frequency,
                                    String remoteClientId,
                                    Long pointNumber) {
        this.dataBus = dataBus;
        this.measurementService = measurementService;
        this.stationCode = stationCode;
        this.timeOut = timeOut * 1000;
        this.frequency = frequency * 1000;
        this.pollSize = pollSize;
        this.pointNumber = pointNumber;
        this.remoteClientId = remoteClientId;
    }

    @Override
    public void toSend(Measurements measurements) {
        //发布云端消息
        if (!measurements.getMeasurementList().isEmpty()) {
            logger.info("发送云端消息{}", Json.encode(measurements.getMeasurementList()));
            logger.info("发送云端消息条数{}", measurements.getMeasurementList().size());
            CloudMeasureValues cloudMeasureValues1 = new CloudMeasureValues();
            List<CloudMeasureValue> collect = measurements.getMeasurementList()
                    .stream().map(Measurement::toCloudMeasureValue).collect(Collectors.toList());
            cloudMeasureValues1.setValuesList(collect);
            cloudMeasureValues1.setProjectCode(stationCode);
            byte[] payload = MessageKit.toByteArray(cloudMeasureValues1);
            dataBus.publish(MqttTopicConstants.MEAS_UPLOAD.value(stationCode), payload, 1);
        }
    }

    @Override
    public void heartBeatStart() {
        dataBus.subscribe("$SYS/broker/connection/" + remoteClientId +  "/state", 1,ThreadMode.Async,  (beeMessage2) -> {
            byte[] bytes = beeMessage2.getM().getPayload().getValue().toByteArray();
            String result = new String(bytes);
            if ("1".equals(result)) {
                logger.info("桥接broker连接成功");
                MqttRemoteMeasurePublish.bridgeBrokerConnected.set(true);
            } else {
                logger.warn("桥接broker连接失败({})", result);
                MqttRemoteMeasurePublish.bridgeBrokerConnected.set(false);
            }
            return null;
        });
        dataBus.subscribe(MqttTopicConstants.HEARTBEAT_SERVER.value(stationCode), 1, ThreadMode.Async, (beeMessage2) -> {
            byte[] bytes = beeMessage2.getM().getPayload().getValue().toByteArray();
            logger.info("收到心跳连接：{},信息{}", MqttTopicConstants.HEARTBEAT_CLIENT.value(stationCode),new String(bytes));
            systemHeartBeatTime.set(Instant.now().toEpochMilli());
            return null;
        });
        vertx.setPeriodic(frequency, a -> {
            String topic3 = MqttTopicConstants.HEARTBEAT_CLIENT.value(stationCode);
            HashMap<String, Object> parms = new HashMap<>();
            parms.put("hearbeat", Instant.now().toEpochMilli());
            try {
                String s = objectMapper.writeValueAsString(parms);
                logger.info("发送心跳信息 topic {},信息{}", topic3, s);
                dataBus.publish(topic3, s, 1);
            } catch (JsonProcessingException e) {
                logger.error("",e);
            }
        });
    }

    @Override
    public void resendStart() {
        dataBus.subscribe(MqttTopicConstants.RECOLLECT_ORDER.value(stationCode), 0, ThreadMode.Async, (beeMessage2) -> {
            String incomingMessage = new String(beeMessage2.getM().getPayload().getValue().toByteArray());
            logger.warn("接受到量测重传指令{}", incomingMessage);
            try {
                DataTransferDTO dataTransferDTO = objectMapper.readValue(incomingMessage, DataTransferDTO.class);
                resent(this.dataBus, dataTransferDTO);
            } catch (JsonProcessingException e) {
                logger.error("消息序列化失败",e);
            }
            return null;
        });
        vertx.setPeriodic(frequency * 12, time -> {
            if (this.isConnect()) {
                LocalDateTime localDateTime = LocalDateTime.now();
                this.resent(this.dataBus,localDateTime);
            }
        });
    }

    /**
     * 被动从备份表中数据进行上传
     * @param dataBus
     * @param message
     */
    public  void resent(MqttDataBus dataBus, DataTransferDTO message) {
        try {
            logger.info("开始被动检查MQTT断线数据时间{}-{}",message.getDtime().get(0),message.getDtime().get(1));
            Long total = measurementService.findMeasurementBackAllSize(message.getDtime().get(0),message.getDtime().get(1));
            logger.info("开始补发消息量测,进行补发操作代发数量{}",total);
            if (total != null && total > 0) {
                long v = total / pollSize + 1;
                for (int i = 0; i <= v; i++) {
                    if (!isConnect()) {
                        logger.info("MQTT网络断线无法进行重发");
                        return;
                    }
                    List<Measurement> all = measurementService.findMeasurementBackAll(message.getDtime().get(0),message.getDtime().get(1),  pollSize, i * pollSize);
                    if (all != null && !all.isEmpty()) {
                        CloudMeasureValues cloudMeasureValues = toCloudMeasureValues(all);
                        byte[] bytes = MessageKit.toByteArray(cloudMeasureValues);
                        logger.warn("重传量测数据第{}次,传输条数{},总条数{}", i, all.size(),total);
                        dataBus.publish(MqttTopicConstants.RECOLLECT.value(stationCode), bytes, 1);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("点号数据重传消息处理失败 错误原因", e);
        }
    }

    /**
     * 主动检查临时缓存表中是否有数据有数据就上传
     * @param dataBus
     * @param time
     */
    public  void resent(MqttDataBus dataBus,LocalDateTime time) {
        try {
            logger.info("开始主动检查MQTT断线数据");
            Long total = measurementService.findMeasurementTempAllSize(time);
            logger.info("开始补发消息量测,进行补发操作代发数量{}",total);
            if (total != null && total > 0) {
                long v = total / pollSize + 1;
                for (int i = 0; i <= v; i++) {
                    if (!isConnect()) {
                        logger.info("MQTT网络断线无法进行重发");
                        return;
                    }
                    List<Measurement> all = measurementService.findTempAll(time, pollSize,i * pollSize);
                    if (all != null && !all.isEmpty()) {
                        int finalI = i;
                        CloudMeasureValues cloudMeasureValues = toCloudMeasureValues(all);
                        byte[] bytes = MessageKit.toByteArray(cloudMeasureValues);
                        logger.warn("重传量测数据第{}次,传输条数{},总条数{}", finalI, all.size(), total);
                        dataBus.publish(MqttTopicConstants.RECOLLECT.value(stationCode), bytes, 1);
                    }
                }
                measurementService.deleteTempMeausremnet(time);
            }
        } catch (Exception e) {
            logger.error("点号数据重传消息处理失败 错误原因", e);
        }
    }


    @Override
    public void onTempStore(Measurements all) {
        measurementService.insertTempMeasurement(all.getMeasurementList());
    }

    @Override
    public boolean isConnect() {
//        return ((Instant.now().toEpochMilli() - systemHeartBeatTime.get()) < timeOut) && bridgeBrokerConnected.get();
        return bridgeBrokerConnected.get();
    }

    @Override
    public ProtocolConstants getProtocol() {
        return protocol;
    }

    @Override
    public MeasurementService getMeasurementService() {
        return measurementService;
    }

    /**
     *
     * 转换成proto消息
     * @return
     */
    public  CloudMeasureValues toCloudMeasureValues(List<Measurement> measurements) {
        CloudMeasureValues cloudMeasureValues = new CloudMeasureValues(stationCode);
        List<CloudMeasureValue> values = new ArrayList<>(measurements.size());
        cloudMeasureValues.setValuesList(values);
        for (int i = 0; i < measurements.size(); i++) {
            Measurement po = measurements.get(i);
            if (po != null) {
                values.add(po.toCloudMeasureValue());
            }
        }
        return cloudMeasureValues;
    }
}
